package com.gmall.app


import java.time.{Instant, LocalDateTime, ZoneId}
import java.time.format.DateTimeFormatter

import com.alibaba.fastjson.JSON
import com.gmall.bean.StartupLog
import com.gmall.utils.{Constants, MyKafkaUtil, MyRedisUtil, PropertiesUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * <p>项目名称: 实时数仓项目 </p>
 * <p>文件名称: 统计当日活跃用户 </p>
 * <p>描述: 日活总数、日活分时统计 </p>
 * <p>创建时间: 2020/10/15</p>
 * @author <a href="mail to: 1034652894@qq.com" rel="nofollow">Demik Qin</a>
 * @version v1.0
 * @update [序号][日期YYYY-MM-DD] [更改人姓名][变更描述]
 */
object DauApp {

    private val zkUrl: String = PropertiesUtil.getValue("zkUrl")


    def main(args: Array[String]): Unit = {
        // 通过流的方式去消费数据
        // 1. 创建一个 StreamingContext
        val conf: SparkConf = new SparkConf().setAppName("DauApp").setMaster("local[*]")
        val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
        // 2. 从kafka获取流
        val sourceStream = MyKafkaUtil.getKafkaStream(ssc, "DauApp", Constants.STARTUP_TOPIC)
        // 3. 对流做各种转换操,  输出(print/foreachPartitions)

        // 3.1 对数据做封装, 封装到样例类中
        // val startupLogStream = sourceStream.map(jsonString => JSON.parseObject(jsonString, classOf[StartupLog]))
        val startupLogStream2 = sourceStream.map(jsonString => {
            val log = JSON.parseObject(jsonString, classOf[StartupLog])
            val ldt = LocalDateTime.ofInstant(Instant.ofEpochMilli(log.ts), ZoneId.systemDefault())
            val yyyyMMddFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd")
            val HHFormat = DateTimeFormatter.ofPattern("HH")
            log.logDate = ldt.format(yyyyMMddFormat)
            //log.logDate = "2020-10-15"
            log.logHour = ldt.format(HHFormat)
            log
        })
        // 3.2 保留第一次启动记录，使用redis来存储
        /*
           采用分区处理算子，以分区为单位创建连接，这样不用每条数据都创建一个连接，
                           可以提高效率，若数据量较大，每条数据都创建一个连接，可能造成redis宕机
        */
        val filteredStartupLogStream = startupLogStream2.mapPartitions(iter => {
            // 1. 获取redis的客户端，不能写在算子外面，会形成闭包，但client连接不支持序列化
            val client = MyRedisUtil.getJedisClient
            // 2. 分区过滤，此filter是scala的方法，并不是RDD算子
            val result = iter.filter(log => {
                // 3. 写入到set: 返回1表示第一次, 返回0则不是第一次，将返回值是1的log保留
                // 以dau:+日期为key，value就是设备id
                client.sadd(s"dau:${log.logDate}", log.mid) == 1
            })
            // 4. 关闭客户机
            client.close()
            // 5 .返回过滤后的log日志
            result
        })
          .foreachRDD(
              rdd => {
              println("-------------------------------------------")
              println(s"Time: ${System.currentTimeMillis()}")
              println("-------------------------------------------")
              import org.apache.phoenix.spark._
              rdd.saveToPhoenix(
                  "GMALL_DAU",
                  Seq("MID", "UID", "APPID", "AREA", "OS", "CHANNEL", "LOGTYPE", "VERSION", "TS", "LOGDATE", "LOGHOUR"),
                  zkUrl = Option(zkUrl))
          })



        // 3. 启动上下文
        ssc.start()
        // 5. 阻止主线程退出, 防止流关闭
        ssc.awaitTermination()
    }

}
